package com.tcm.kafka;

import org.apache.kafka.common.serialization.Deserializer;

import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.util.Map;

public class CompanyDeserializer implements Deserializer<Company> {
    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {

    }

    @Override
    public Company deserialize(String topic, byte[] bytes) {
        if(bytes.length == 0) return null;

        ByteBuffer buffer = ByteBuffer.wrap(bytes);
        int nameLen = buffer.getInt();
        byte[] names = new byte[nameLen];
        buffer.get(names);
        int addressLen = buffer.getInt();
        byte[] address = new byte[addressLen];
        buffer.get(address);

        String name=null,addr=null;
        try {
            name = new String(names, "UTF-8");
            addr = new String(address, "UTF-8");
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }

        return new Company(name, addr);
    }

    @Override
    public void close() {

    }
}
